package com.wuercloud.mq.message.sender;

import com.rabbitmq.client.Channel;
import com.wuercloud.mq.common.MessageConverterUtils;
import com.wuercloud.mq.message.sender.cache.MessageFaultCache;
import com.wuercloud.mq.common.MqConstants;
import com.wuercloud.mq.common.ResultRes;
import com.wuercloud.mq.message.converter.FastJsonMessageConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created by JiangWL on 2017/3/23.
 */
public class WawaMQSendBuilder {
    private static Logger logger = LoggerFactory.getLogger(WawaMQSendBuilder.class);
    private MessageFaultCache faultCache;
    public WawaMQSendBuilder(MessageFaultCache faultCache) {
        this.faultCache = faultCache;
    }

    /**
     * 构建消息发送器
     * @param queueName
     * @return
     */
    public WawaMQSender buildSender(final  String queueName,final RabbitTemplate rabbitTemplate){

        try {
            buildQueue("520WAWA-".concat(queueName.substring(0,queueName.lastIndexOf("-"))), queueName, queueName, rabbitTemplate.getConnectionFactory().createConnection());
        } catch (IOException e) {
            logger.error("构建队列出错。");
            e.printStackTrace();
        }

        final RabbitTemplate newRabbitTemplate = new RabbitTemplate(rabbitTemplate.getConnectionFactory());

        newRabbitTemplate.setMandatory(true);
        newRabbitTemplate.setExchange("520WAWA-".concat(queueName.substring(0,queueName.lastIndexOf("-"))));
        newRabbitTemplate.setRoutingKey(queueName);
        newRabbitTemplate.setMessageConverter(new FastJsonMessageConverter());

        //消息发送确认返回信息
        newRabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                logger.info("消息确认成功" + correlationData.toString());
                //消息收到删除缓存信息
                faultCache.del(correlationData.getId());
            } else {
                //处理丢失的消息（nack）
                logger.info("消息确认失败");
            }
        });
        //此与Mandatory相关连，Mandatory必须为true
        newRabbitTemplate.setReturnCallback((msg, replyCode, replyText, tmpExchange, tmpRoutingKey) -> {
            try {
                Thread.sleep(MqConstants.ONE_SECOND);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            logger.info("send message failed: " + replyCode + " " + replyText);
            newRabbitTemplate.convertAndSend(queueName,msg);
        });
        return new WawaMQSender() {
            @Override
            public ResultRes sendMessage(final Object message,final String expiration) {
                try {
                    String id = faultCache.generateId();
                    //发送前先缓存信息，保证数据不丢失，发送成功后删除
                    faultCache.add(id, message,queueName,expiration,this);
                    final Message msg = MessageConverterUtils.convertMessageIfNecessary(message);
                    //设置消息存活时间。从传过来消息内容获取，此处模拟存活30秒
                    msg.getMessageProperties().setExpiration(expiration);
                    newRabbitTemplate.send("520WAWA-".concat(queueName.substring(0,queueName.lastIndexOf("-"))),queueName,msg,new CorrelationData(id));
                    //rabbitTemplate.correlationConvertAndSend(message, new CorrelationData(id));
                } catch (Exception e) {
                    return new ResultRes(false, "N");
                }
                return new ResultRes(true, "Y");
            }
        };
    }

    private void buildQueue(String exchange, String routingKey,
                            final String queue, Connection connection) throws IOException {
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(exchange, "direct", true, false, null);
        channel.basicQos(1);
        channel.queueDeclare(queue, true, false, false, null);
        channel.queueBind(queue, exchange, routingKey);

        try {
            channel.close();
        } catch (TimeoutException e) {
            e.printStackTrace();
            logger.info("close channel time out " + e);
        } finally {
            try {
                channel.close();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
            connection.close();
        }
    }
}
